package com.group1.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.group1.realtime.app.BaseAppV2;
import com.group1.realtime.bean.UserNewAndActiveBean;
import com.group1.realtime.util.AtguiguUtil;
import com.group1.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

import static com.group1.realtime.common.Constant.*;

public class Dws_UserNewAndActive extends BaseAppV2 {
    public static void main(String[] args) {
        new Dws_UserNewAndActive().init(
                3022,
                2,
                "Dws_UserNewAndActive",
                TOPIC_DWD_TRAFFIC_PAGE,
                TOPIC_DWD_USER_NEW_ADD
        );
    }



    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> stream) {


        DataStream<UserNewAndActiveBean> unionStream = processStream(stream);

        SingleOutputStreamOperator<UserNewAndActiveBean> aggregatedStream = windowAndAgg(unionStream);
        aggregatedStream.print();
        // 3. 把结果写入到clickHouse中
        writeToClickHouse(aggregatedStream);
    }

    private void writeToClickHouse(SingleOutputStreamOperator<UserNewAndActiveBean> aggregatedStream) {
        aggregatedStream.addSink(FlinkSinkUtil.getClickHouseSink("dws_user_new_active_window", UserNewAndActiveBean.class));
    }

    private SingleOutputStreamOperator<UserNewAndActiveBean> windowAndAgg(DataStream<UserNewAndActiveBean> unionStream) {

        return unionStream.assignTimestampsAndWatermarks(
                WatermarkStrategy
                        .<UserNewAndActiveBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                        .withIdleness(Duration.ofSeconds(5))
                ).windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<UserNewAndActiveBean>("late") {})
                .reduce(
                        new ReduceFunction<UserNewAndActiveBean>() {
                            @Override
                            public UserNewAndActiveBean reduce(UserNewAndActiveBean value1, UserNewAndActiveBean value2) throws Exception {
                                value1.setActiveCt(value1.getActiveCt() + value2.getActiveCt());
                                value1.setNewAddCt(value1.getNewAddCt() + value2.getNewAddCt());
                                return value1;
                            }
                        },
                        new ProcessAllWindowFunction<UserNewAndActiveBean, UserNewAndActiveBean, TimeWindow>() {
                            @Override
                            public void process(Context ctx, Iterable<UserNewAndActiveBean> elements, Collector<UserNewAndActiveBean> collector) throws Exception {

                                UserNewAndActiveBean bean = elements.iterator().next();
                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setTs(System.currentTimeMillis());
                                collector.collect(bean);
                            }
                        }
                );
    }

    private DataStream<UserNewAndActiveBean> processStream(Map<String, DataStreamSource<String>> stream) {

        SingleOutputStreamOperator<UserNewAndActiveBean> activeStream = stream.get(TOPIC_DWD_TRAFFIC_PAGE)
                .map(JSON::parseObject)
                .keyBy(object -> object.getJSONObject("common").getString("uid"))
                .process(new KeyedProcessFunction<String, JSONObject, UserNewAndActiveBean>() {

                    private ValueState<Long> loginState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        loginState = getRuntimeContext().getState(new ValueStateDescriptor<Long>("loginState", Long.class));
                    }

                    @Override
                    public void processElement(JSONObject object, Context context, Collector<UserNewAndActiveBean> collector) throws Exception {

                        Long ts = object.getLong("ts");
                        if (loginState.value() == null) {
                            loginState.update(ts);
                        } else if (ts - loginState.value() < 3600 * 24 * 2 * 1000) {
                            collector.collect(new UserNewAndActiveBean("", "", 0L,1L, ts));
                        }
                        loginState.update(ts);
                    }
                });

        SingleOutputStreamOperator<UserNewAndActiveBean> newAddStream = stream.get(TOPIC_DWD_USER_NEW_ADD)
                .map(new MapFunction<String, UserNewAndActiveBean>() {
                    @Override
                    public UserNewAndActiveBean map(String value) throws Exception {
                        JSONObject obj = JSON.parseObject(value);
                        System.out.println(obj.toJSONString());
                        Long ts = obj.getLong("ts") * 1000;
                        return new UserNewAndActiveBean("","",1L,0L,ts);
                    }
                });
        newAddStream.print();

        return newAddStream.union(activeStream);

    }


}
